package com.neuedu.brazil;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;

/**
 * 多部算法：统计每年的天气
 *
 * @author DEll
 */
public class Step3 {
    public static void run(String input, String output) {
        try {
            //定义输入目录;
            // String input = "/brazil_weather";
            // String output = "/step1_output";
            Path outputPath = new Path(output);
            // 判断输出目录是否存在，存在则删除之
            HadoopUtils.delete(outputPath);
            // 创建Job任务
            Job job = Job.getInstance(HadoopUtils.getConf(), "统计每年的天气");
            //设置输入
            job.setInputFormatClass(TextInputFormat.class);
            FileInputFormat.setInputPaths(job, input);
            // 设置mapper
            job.setMapperClass(Step3Mapper.class);
            job.setMapOutputKeyClass(Text.class);
            job.setMapOutputValueClass(WeatherWritable.class);
            //设置reducer
            job.setReducerClass(Step3Reducer.class);
            // 设置输出:下面2行容易写错
            job.setOutputKeyClass(WeatherWritable.class);
            job.setOutputValueClass(NullWritable.class);
            job.setOutputFormatClass(TextOutputFormat.class);
            FileOutputFormat.setOutputPath(job, outputPath);
            // 运行
            boolean flag = job.waitForCompletion(true);
            if (flag) {
                System.out.println("统计汇总每年的天数，结束~~~");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private static class Step3Mapper extends Mapper<LongWritable, Text, Text, WeatherWritable> {
        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, WeatherWritable>.Context context) throws IOException, InterruptedException {
            //读取数据
            //83377 01/01/1963 0.0 29.0   16.7   21.74
            //83377 01/01/1964 3.2 26.0   18.0   20.84
            //83377 01/01/1965 21.2     24.7    16.6    19.66
            String line = value.toString();
            if (StringUtils.isBlank(line)) {
                return;
            }
            String[] items = line.split("\t");
            if (items.length != 6) {
                return;
            }
            //提取数据
            String code = items[0];
            String date = items[1].substring(6);
            double precipitation = Double.parseDouble(items[2]);
            double maxTemperature = Double.parseDouble(items[3]);
            double minTemperature = Double.parseDouble(items[4]);
            double avgTemperature = Double.parseDouble(items[5]);
            WeatherWritable w = new WeatherWritable(code, date, precipitation, maxTemperature, minTemperature, avgTemperature);
            //发送数据
            context.write(new Text(code + ";" + date), w);
        }

    }

    private static class Step3Reducer extends Reducer<Text, WeatherWritable, WeatherWritable, NullWritable> {
        @Override
        protected void reduce(Text key, Iterable<WeatherWritable> values, Reducer<Text, WeatherWritable,
                WeatherWritable, NullWritable>.Context context) throws IOException, InterruptedException {
            String code = key.toString().split(";")[0];
            String date = key.toString().split(";")[1];
            WeatherWritable w = new WeatherWritable(code, date);
            w.setMinTemperature(Double.MAX_VALUE);
            int count = 0;
            for (WeatherWritable value : values) {
                w.setPrecipitation(w.getPrecipitation() + value.getPrecipitation());
                w.setMaxTemperature(Math.max(w.getMaxTemperature(), value.getMaxTemperature()));
                w.setMinTemperature(Math.min(w.getMinTemperature(), value.getMinTemperature()));
                w.setAvgTemperature(w.getAvgTemperature() + value.getAvgTemperature());
                count++;
            }
            w.setAvgTemperature(w.getAvgTemperature() / count);
            context.write(w, NullWritable.get());
        }
    }
}


